package mspbots.xero.data.core.synclogs;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.xero.data.core.AbstractToolsUtil;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * mspbots.data.sync.autotask.badmessage.BadMessageReceiver
 *
 * @author <a href="https://github.com/vnobo">Alex bob</a>
 * @date Created by 2020/7/21
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class BadMsgReceiver extends AbstractToolsUtil {

    private final BadMsgRepository messageRepository;

    @SuppressWarnings("unchecked")
    @RabbitListener(queues = "xero.bad.message")
    public void syncTask(Message<Object> message, @Headers Map<String, Object> headers) {

        try {

            JsonNode msg = super.objectMapper.convertValue(message.getPayload(), JsonNode.class);
            long timestamp = Long.parseLong(headers.get("timestamp").toString());
            Instant instant = Instant.ofEpochMilli(timestamp);

            List<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get("x-death");
            Map<String, ?> xDeathMap = xDeath.get(0);
            Long count = (Long) xDeathMap.get("count");
            Date time = (Date) xDeathMap.get("time");
            List<String> keys = (List<String>) xDeathMap.get("routing-keys");

            BadMsg badMessage = BadMsg.builder()
                    .tenantId(msg.get("tenantId").asLong(0L))
                    .msgId(headers.get("id").toString())
                    .msg(msg)
                    .count(count.intValue())
                    .exchange(xDeathMap.get("exchange").toString())
                    .queue(xDeathMap.get("queue").toString())
                    .reason(xDeathMap.get("reason").toString())
                    .routingKeys(StringUtils.collectionToCommaDelimitedString(keys))
                    .errTime(LocalDateTime.ofInstant(time.toInstant(), ZoneId.systemDefault()))
                    .msgTime(LocalDateTime.ofInstant(instant, ZoneId.systemDefault()))
                    .build();

            this.messageRepository.save(badMessage).block(Duration.ofSeconds(10));

        } catch (Exception e) {
            log.error("Bad msg receiver error.");
            log.error("Sync bad message body {}", message.getPayload());
            log.error("Sync bad message headers {}", headers);
        }
    }

}
